package com.fanshuai.spring.boot.kafkaTransaction.service;

import com.fanshuai.spring.boot.kafkaTransaction.consuner.handler.KafkaHandler;
import com.fanshuai.spring.boot.kafkaTransaction.consuner.listener.ConsumerListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Objects;

@ConditionalOnProperty(value = "spring.kafka.enabled", matchIfMissing = false)
@Service
@Slf4j
public class KafkaConsumerService {
    @Autowired(required = false)
    private List<KafkaHandler> handlerList;
    @Autowired(required = false)
    private List<ConsumerListener> listeners;

    @KafkaListener(topics = {"topic1"})
    public void processTopic1(ConsumerRecord<String, String> consumerRecord) {
        basicConsume(consumerRecord);
    }

    @KafkaListener(topics = {"topic2"})
    public void processTopic2(ConsumerRecord<String, String> consumerRecord) {
        basicConsume(consumerRecord);
    }

    private void basicConsume(ConsumerRecord<String, String> consumerRecord) {
        List<ConsumerListener> customListeners = new ArrayList<>();
        if (!CollectionUtils.isEmpty(listeners)) {
            customListeners.addAll(listeners);
            customListeners.sort(Comparator.comparingInt(ConsumerListener::order));

            customListeners.forEach(listener -> {
                try {
                    listener.beforeConsume(consumerRecord);
                } catch (Exception e) {
                    log.error("before consume error, ", e);
                }
            });
        }
        String topic = consumerRecord.topic();
        for (KafkaHandler handler : handlerList) {
            if (Objects.equals(topic, handler.topic())) {
                try {
                    handler.handle(consumerRecord.key());
                } catch (Exception e) {
                    log.error("consume error, ", e);
                }
                break;
            }
        }

        if (!CollectionUtils.isEmpty(customListeners)) {
            customListeners.forEach(listener -> {
                try {
                    listener.afterConsume(consumerRecord);
                } catch (Exception e) {
                    log.error("consume error, ", e);
                }
            });
        }
    }
}
